package zybMQ

import (
	"fmt"
	"github.com/golang/protobuf/proto"
	"github.com/streadway/amqp"
	"os"
	"sync"
	"time"
	"unsafe"
)

// 定义订阅主题结构
type stQueueData struct {
	queueName  string // 队列名称
	routingKey string // key名称
	channel    *amqp.Channel
}

// 定义RabbitMQ对象
type RabbitMQ struct {
	conn          *amqp.Connection
	senderchannel *amqp.Channel
	serverName    string                  // 服务名称
	exchangeName  string                  // 交换机名称:entry
	exchangeType  string                  // 交换机类型:topic
	rabbitUrl     string                  // 地址
	queueList     []stQueueData           // 订阅的主题
	on_message    func(dataByte []byte)   // 消息处理函数
	on_messageEx  func(msg amqp.Delivery) // 消息处理函数
	bAutoAck      bool
	bExit         bool
	mu            sync.RWMutex
}

// auto ack message.
func NewMQ(svrName, mqUrl string, f func(dataByte []byte)) *RabbitMQ {
	return &RabbitMQ{
		serverName:   svrName,
		rabbitUrl:    mqUrl,
		on_message:   f,
		exchangeName: "entry",
		exchangeType: "topic",
		bAutoAck:     true,
	}
}

// need manual ack message.
func NewMQEx(svrName, mqUrl string, fx func(msg amqp.Delivery)) *RabbitMQ {
	return &RabbitMQ{
		serverName:   svrName,
		rabbitUrl:    mqUrl,
		on_messageEx: fx,
		exchangeName: "entry",
		exchangeType: "topic",
		bAutoAck:     false,
		bExit: false,
	}
}

// auto ack.
func (r *RabbitMQ)RegMsgHandle(f func(dataByte []byte))  {
	r.bAutoAck = true
	r.on_message = f
}

// not auto ack.
func (r *RabbitMQ)RegMsgHandleEx(fx func(msg amqp.Delivery))  {
	r.bAutoAck = false
	r.on_messageEx = fx
}

func (r *RabbitMQ) SubscribTopic(topic string) {
	tmp := stQueueData{
		queueName:  r.serverName + "_" + topic,
		routingKey: topic,
	}

	r.queueList = append(r.queueList, tmp)
}

func checkerror(strlabel string, err error) {
	if err != nil {
		fmt.Println(strlabel, "fail :", err)
		os.Exit(2)
	}
}

func (r *RabbitMQ)doReconnect(c chan *amqp.Error)  {
	e := <-c

	if r.bExit {
		return
	}

	fmt.Print("\n---------amqp error----------\n")
	fmt.Println(e)
	time.Sleep(1*time.Second)
	for i := 0; i < 100; i++ {
		r.Start()

		if r.conn == nil || r.conn.IsClosed() {
			time.Sleep(1*time.Second)
			continue
		}else {
			break
		}
	}
}

func (r *RabbitMQ) Start() {
	// connect mq.
	{
		conn, err := amqp.Dial(r.rabbitUrl)
		checkerror("connect mq", err)
		r.conn = conn
	}

	// for reconnect mq.
	{
		c := make(chan *amqp.Error)
		r.conn.NotifyClose(c)
		go r.doReconnect(c)
	}

	// declare sender channel.
	{
		var err error
		r.senderchannel, err = r.conn.Channel()
		checkerror("declare sender channel", err)
		err = r.senderchannel.ExchangeDeclare(r.exchangeName,
			r.exchangeType, false, false, false, false, nil)
		checkerror("declare sender exchange", err)
	}

	if r.on_message == nil && r.on_messageEx == nil{
		fmt.Println("no msseage deal func.")
		os.Exit(1)
	}

	// bind receiver channel & queue.
	for i := 0; i < len(r.queueList); i++ {
		var err error
		r.queueList[i].channel, err = r.conn.Channel()
		checkerror("delcare receiver channel", err)
		err = r.queueList[i].channel.ExchangeDeclare(r.exchangeName,
			r.exchangeType, false, false, false, false, nil)
		checkerror("delcare receiver exchange", err)
		var q amqp.Queue
		q, err = r.queueList[i].channel.QueueDeclare(r.queueList[i].queueName,
			true, false, false, false, nil)
		checkerror("declare queue", err)
		err = r.queueList[i].channel.QueueBind(q.Name, r.queueList[i].routingKey,
			r.exchangeName, true, nil)
		checkerror("bind queue", err)
		// consume msg.
		if r.bAutoAck {
			go func(queue stQueueData, f func(dataByte []byte)) {
				msgList, err := queue.channel.Consume(queue.queueName,
					"", false, false, false, false, nil)
				checkerror("consume msg", err)
				for msg := range msgList {
					f(msg.Body)
					err := msg.Ack(true)
					if err != nil {
						fmt.Println("ack message fail:", err)
					}
				}
			}(r.queueList[i], r.on_message)
		} else {
			go func(queue stQueueData, f func(msg amqp.Delivery)) {
				msgList, err := queue.channel.Consume(queue.queueName,
					"", false, false, false, false, nil)
				checkerror("consume msg", err)
				for msg := range msgList {
					f(msg)
				}
			}(r.queueList[i], r.on_messageEx)
		}
	}
}

func (r *RabbitMQ) Stop() {
	if r.conn == nil {
		return
	}

	r.bExit = true

	err := r.senderchannel.Close()
	checkerror("close senderchannel", err)

	for i := 0; i < len(r.queueList); i++ {

		_, err = r.queueList[i].channel.QueueDelete(r.queueList[i].queueName, false, false, true)
		if err != nil {
			fmt.Println("queuedelete fail. queue name:", r.queueList[i].queueName, "error:", err)
		}

		err := r.queueList[i].channel.Close()
		checkerror("close receiver channel", err)
	}

	err = r.conn.Close()
	checkerror("close mq conn", err)
}

func (r *RabbitMQ) PublishMessage(topic string, funCode uint32, sid uint32, data proto.Message) {
	msg, err := proto.Marshal(data)
	if err != nil {
		fmt.Println("Marshal msg fail:", err)
		return
	}

	var sendBuf []byte = make([]byte, len(msg)+8)

	// 赋值funcode
	{
		p := unsafe.Pointer(&funCode)
		q := (*[4]byte)(p)
		copy(sendBuf[0:], (*q)[0:])
	}

	// 赋值sid
	{
		p := unsafe.Pointer(&sid)
		q := (*[4]byte)(p)
		copy(sendBuf[4:], (*q)[0:])
	}

	copy(sendBuf[8:], msg)
	err = r.senderchannel.Publish(r.exchangeName,
		topic, false, false, amqp.Publishing{
			ContentType: "text/plain",
			Body:        sendBuf,
		})

	if err != nil {
		fmt.Printf("MQ任务发送失败:%s \n", err)
		return
	}

	fmt.Printf("send message, funCode[%d] sid[%d] \n %s\n", funCode, sid, data.String())
}
